package com.sdp.irrigation.kafka.task;

import java.util.concurrent.TimeUnit;

import com.sdp.core.bizc.redis.RedisUtil;
import com.sdp.irrigation.cache.IrrigationCacheKeys;
import com.sdp.irrigation.kafka.producer.KafkaProducer;
import com.sdp.irrigation.kafka.property.ServerProperty;
import com.sdp.irrigation.kafka.utils.ProtocolUtils;
import com.sdp.irrigation.timer.NettyTimer;
import com.sdp.irrigation.utils.CommConstants;
import com.sdp.irrigation.utils.JsonUtils;
import com.sdp.irrigation.utils.ProtocolTypeEnum;

import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class EqStatusQueryTask implements TimerTask {
	
	private RedisUtil redisUtil;
	
	private String devSn;
	
	private KafkaProducer kafkaProducer;
	
	private ServerProperty serverProperty;
	
	private static final int MAX_QUERY_COUNT = 3;
	
	private static final long QUERY_TIME_OUT = 60;
	
	public static final int QUERY_DELAY = 3;
	
	
	public EqStatusQueryTask(RedisUtil redisUtil, String devSn, KafkaProducer kafkaProducer,
			ServerProperty serverProperty) {
		this.redisUtil = redisUtil;
		this.devSn = devSn;
		this.kafkaProducer = kafkaProducer;
		this.serverProperty = serverProperty;
	}

	public boolean isOpenTask(String devSn) {
		boolean result = false;
		if (!redisUtil.hasKey(IrrigationCacheKeys.getEqStatusQueryCacheKey(devSn))) {
			result = true;
		} else {
			Object value = redisUtil.get(IrrigationCacheKeys.getEqStatusQueryCacheKey(devSn));
			if (value != null) {
				int count = (Integer) value;
				if (count <= 0) {
					result = true;
				}
			}
		}
		redisUtil.set(IrrigationCacheKeys.getEqStatusQueryCacheKey(devSn), MAX_QUERY_COUNT, QUERY_TIME_OUT);
		return result;
	}

	@Override
	public void run(Timeout timeout) throws Exception {
		int count = 0;
		
		try {
			Object value = redisUtil.get(IrrigationCacheKeys.getEqStatusQueryCacheKey(devSn));
			if (value != null) {
				count = (Integer) value;
				if (count > 0) {
					String topic = ProtocolUtils.getDownMessageTopic(ProtocolTypeEnum.REPORT.getCode(), serverProperty.getDownTopicPrefix());
					String msg = JsonUtils.toJSONString(ProtocolUtils.getDownReportCommPackage(devSn));
					kafkaProducer.send(topic, msg);
					
					redisUtil.decrement(IrrigationCacheKeys.getEqStatusQueryCacheKey(devSn));
				}
			}
			
		} catch (Exception e) {
			CommConstants.printErrorLog(e, log);
		} finally {
			if (count > 0) {
				NettyTimer.newTimeout(timeout.task(), QUERY_DELAY, TimeUnit.SECONDS);
			}
		}
	}

}
